/* <pre>
 * Copyright (c) The Loose Coupling Team.
 * Licensed under the Apache License, Version 2.0 (the "License"); 
 * you may not use this file except in compliance with the License. 
 * You may obtain a copy of the License at 
 *
 *      http://www.apache.org/licenses/LICENSE-2.0 
 *
 * Unless required by applicable law or agreed to in writing, software 
 * distributed under the License is distributed on an "AS IS" BASIS, 
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
 * See the License for the specific language governing permissions and 
 * limitations under the License.
 *</pre>
 */
package org.lc.eic.synchronizer.request;

import java.sql.Timestamp;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

import org.apache.log4j.Logger;
import org.lc.eic.api.requestreply.exception.RequestProcessingException;
import org.lc.eic.api.requestreply.sync.RequestDispatcher;
import org.lc.eic.core.pipesandfilters.Pipe;
import org.lc.eic.core.pipesandfilters.exception.ProcessingException;
import org.lc.eic.synchronizer.CorrelationIdentifier;
import org.lc.eic.synchronizer.ExpirationIdentifier;
import org.lc.eic.synchronizer.UniqueIDIdentifier;
import org.lc.eic.synchronizer.request.exception.CorrelationException;
import org.lc.eic.synchronizer.request.exception.ExpirationException;
import org.lc.eic.synchronizer.request.exception.UniqueIDException;
import org.lc.eic.synchronizer.request.impl.OngoingRequest;
import org.lc.eic.synchronizer.request.impl.OngoingRequestsHolder;



public class RequestSynchronizer<I,O> implements RequestDispatcher<I,O>{
	
	private static final Logger logger = Logger.getLogger(RequestSynchronizer.class);
	
	
	private CorrelationIdentifier<I> requestCorrelationIdentifier;
	private CorrelationIdentifier<O> replyCorrelationIdentifier;	
	private ExpirationIdentifier<I> expirationIdentifier;	
	private UniqueIDIdentifier<I> uniqueIDIdentifier;
	
	

	private Map<String, OngoingRequest<I,O>> ongoingRequests = Collections.synchronizedMap( new HashMap<String, OngoingRequest<I,O>>() );
	
	
	private org.lc.eic.api.requestreply.async.RequestDispatcher<I, O> asyncRequestDispatcher;
	
	private int threadPoolSize = 10;
	
	private ExecutorService threadedService = Executors.newFixedThreadPool(threadPoolSize, new ThreadFactory() {
	
		public Thread newThread(Runnable r) {
			Thread thread = new Thread(r);
			thread.setDaemon(true);
			return thread;
		}
	
	});
	

	/**
	 * @see org.lc.eic.api.requestreply.sync.RequestDispatcher#request(java.lang.Object)
	 */
	public O request(final I request) throws RequestProcessingException {
		
		//Do we have instance
		if(request == null){
			throw new RequestProcessingException("Cannot send null-requests!");
		}
		
		//Get the correlation id.
		String correlationId = null;
		try{
			correlationId = requestCorrelationIdentifier.getCorrelationID(request);
		} catch(CorrelationException e) {
			throw new RequestProcessingException("Failed identify correlation-id! [request="+request.toString()+"]", e);			
		}	
		
		//Check that it is valid
		if(correlationId == null){
			throw new RequestProcessingException("Got null correlation-id! [request="+request.toString()+"]");
		}
		
		//Get a expiration time...
		Timestamp expiration = null;
		
		try {
			expiration = expirationIdentifier.getExpirationTime(request);
		} catch (ExpirationException e) {
			throw new RequestProcessingException("Failed resolve expiration time! [request="+request.toString()+"]", e);
		}
		
		if(expiration == null){
			throw new RequestProcessingException("Null expiration time! [request="+request.toString()+"]");
		}
		
		//Get a uniqueId time...
		String uniqueId = null;
		
		try {
			uniqueId = uniqueIDIdentifier.getUniqueID(request);
		} catch (UniqueIDException e) {
			throw new RequestProcessingException("Failed resolve unique id! [request="+request.toString()+"]", e);
		}
		
		if(uniqueId == null){
			throw new RequestProcessingException("Null unique id time! [request="+request.toString()+"]");
		}
		
		final OngoingRequest<I,O> ongoingRequest = new OngoingRequest<I,O>();
		ongoingRequest.setCorrelationID(correlationId);
		ongoingRequest.setExpirationTime(expiration);
		ongoingRequest.setUniqueID(uniqueId);
		ongoingRequest.setRequest(request);		
		
		synchronized(ongoingRequests){
			if(ongoingRequests.containsKey(correlationId)){
				throw new RequestProcessingException("The correlation id is already in use!! [correlationId="+correlationId+"]");
			}		
			ongoingRequests.put(correlationId, ongoingRequest);	
		}
		
		if(logger.isDebugEnabled()){
			logger.debug("Executing asynchronous request [correlationID="+correlationId+", expirationTime="+expiration.toString()+", request={"+request.toString()+"}]");
		}
		
	
		
		
		Runnable runnable = new Runnable() {
		
			public void run() {
			
				try {
					asyncRequestDispatcher.getRequestPipe().process(request);
				} catch (ProcessingException e) {
					ongoingRequest.cancelRequest("Failed delivering asynchronous message.", e);
				}
		
			}
		
		};
		
		threadedService.execute(runnable);	
		
		
		try{
			if(logger.isDebugEnabled()){
				logger.debug("Waiting for response. [correlationID="+correlationId+"]");
			}
			ongoingRequest.waitForResponse();			
		
		} finally{
			ongoingRequests.remove(correlationId);
		}
		
		
		O reply  = ongoingRequest.getReply();
		
		if(logger.isDebugEnabled()){
			logger.debug("Request with reply returns. [correlationID="+correlationId+", expirationTime="+expiration.toString()+", request={"+request.toString()+"}, reply={"+reply.toString()+"}]");
		}
		
		return reply;
			
		
		
		
	}

	
	//Create a reply pipe
	private Pipe<O> getReplyPipe(){
		return new Pipe<O>() {
			//just delegate to private method below...
			public void process(O reply) throws ProcessingException{
				receiveReply(reply);
			}
		};
	}
	
	/**
	 * Here is where incoming replies land...
	 * 
	 */
	private void receiveReply(O reply) throws ProcessingException {
		if(reply == null){
			throw new ProcessingException("Null reply received!!");
		}
		
		String correlationId = null;
		
		try {
			correlationId = replyCorrelationIdentifier.getCorrelationID(reply);
		} catch (CorrelationException e) {
			throw new ProcessingException("Failed locating correlation id. [replyMessage"+reply.toString()+"]", e);
		}
		
		if(correlationId == null){
			throw new ProcessingException("Got null correlation-id! [request="+reply.toString()+"]");
		}
		
		if(logger.isDebugEnabled()){
			logger.debug("Response arrived. [correlationID="+correlationId+", reply={"+reply.toString()+"}]");
		}
		
		OngoingRequest<I,O> ongoingRequest = ongoingRequests.get(correlationId);
		
		if(ongoingRequest == null){
			logger.warn("No ongoing-request was found for reply. [correlationId="+correlationId+", reply="+reply.toString()+"]");
		} else{			
			ongoingRequest.setReply(reply);			
		}
		
		
	}

	public void setReplyCorrelationIdentifier(
			CorrelationIdentifier<O> replyCorrelationIdentifier) {
		this.replyCorrelationIdentifier = replyCorrelationIdentifier;
	}

	public void setRequestCorrelationIdentifier(
			CorrelationIdentifier<I> requestCorrelationIdentifier) {
		this.requestCorrelationIdentifier = requestCorrelationIdentifier;
	}
	
	public void setUniqueIDIdentifier(UniqueIDIdentifier<I> uniqueIDIdentifier) {
		this.uniqueIDIdentifier = uniqueIDIdentifier;
	}

	public void setAsyncRequestDispatcher(org.lc.eic.api.requestreply.async.RequestDispatcher<I, O> nextApp) {
		this.asyncRequestDispatcher = nextApp;
		this.asyncRequestDispatcher.setReplyPipe(getReplyPipe());
	}
	
	 OngoingRequestsHolder<I, O> getOngoingRequest(){
		 return new OngoingRequestsHolder<I, O>() {
			 public Map<String, OngoingRequest<I,O>> getOngoingRequests(){
				 return RequestSynchronizer.this.getOngoingRequests();
			 }
		};
	 }
	
	public int getNoOngoingRequests(){
		return ongoingRequests.size();
	}

	public void setExpirationIdentifier(ExpirationIdentifier<I> expirationIdentifier) {
		this.expirationIdentifier = expirationIdentifier;
	}

	/**
	 * @see com.ford.vcc.ng.eie.messagerouter.requestflow2msgflow.OngoingRequestsHolder#getOngoingRequests()
	 */
	public Map<String, OngoingRequest<I,O>> getOngoingRequests() {
		return ongoingRequests;
	}

	public void setThreadPoolSize(int threadPoolSize) {
		this.threadPoolSize = threadPoolSize;
		
		threadedService = Executors.newFixedThreadPool(threadPoolSize, new ThreadFactory() {
			
			public Thread newThread(Runnable r) {
				Thread thread = new Thread(r);
				thread.setDaemon(true);
				return thread;
			}
		
		});
	}

	
	
}
